//Reactor模式实际上就是帮我们减少了监控链接到来的成本...
#include <sys/epoll.h>
#include <pthread.h>
#include <vector>
#include <iostream>
#include <sys/types.h>          
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <string.h>
const int ThreadBufferSize = 1024;
const int CoreThreadNum=10;
const int MaxThreadNum=100;


class Connection
{
public:
    int event;
    int socketfd=-1;
};

class Thread
{
public:
    int payload=0;
    pthread_t thread_id=-1;

    //线程也可以添加一个双端任务队列，让管理该队列的线程从一段执行，
    //当有其他线程空闲时可以从另一端获取该线程的任务，以缓解该线程的压力。TODO
};

class ThreadPool
{
private:
    std::mutex mtx;
    std::condition_variable cond;
    std::queue<Connection> connections;
    
    int core_thread_num=CoreThreadNum;
    int max_thread_num=MaxThreadNum;
    long long total_load=0; //我想设置一个高水位线，用来警告过量的链接负载，TODO
    std::vector<Thread> threadpool;
    bool is_running=true;

    static void *ThreadLoop(void* self)
    {
        ThreadPool* pointer=(ThreadPool*)self;
        char buffer[ThreadBufferSize];
        while(pointer->is_running)
        {
            if(pointer->connections.size()==0)
            {
                pointer->ThreadWait();
            }
            else 
            {   
                Connection conn=pointer->connections.front();
                pointer->connections.pop();

                //处理请求...       
                //这里细节处理得并不完美，当多次读事件到来，而写事件始终不就绪，
                //就会导致一些问题，这里得代码由于脱离业务怎么写都是存在一定的问题的，
                //所以最好可以把这一部分交给用户处理。TODO
                if((conn.event&EPOLLIN) && strlen(buffer)==0)
                {
                    memset(buffer,0,sizeof(buffer));
                    int num = read(conn.socketfd,buffer,sizeof(buffer)-1);
                }

                if((conn.event&EPOLLOUT) && strlen(buffer)!=0)
                {
                    write(conn.socketfd,buffer,sizeof(buffer));
                }
            }
        }
    }

public:

    ThreadPool()
    {
        while(threadpool.size()<core_thread_num)
        {
            Thread t;
            pthread_create(&t.thread_id,nullptr,ThreadLoop,this);
            pthread_detach(t.thread_id);
            threadpool.push_back(t);
        }
    }

    ~ThreadPool()
    {
        is_running=false;
    }

    void ThreadWait()
    {
        std::unique_lock<std::mutex> lock(mtx);
        cond.wait(lock);
    }
    void AddConnection(Connection conn)
    {
        std::lock_guard<std::mutex> lock(mtx);
        connections.push(conn);
    }

    void NotifyOneThread()
    {
        cond.notify_one();
    }

    void NotifyAllThread()
    {
        cond.notify_all();
    }
};

class Reactor
{
private:
    int efd=-1;
    int listener=-1;
    bool is_running=true;
    ThreadPool threadpool;
public:
    Reactor()
    {
        efd=epoll_create1(0);
        if(-1==efd)
        {
            std::cout<<"epoll create failed! errno is:"<<errno<<std::endl;
        }
    }
    ~Reactor()
    {
        if(listener!=-1)
        {
            close(listener);
        }
        
        if(efd!=-1)
        {
            close(efd);
        }
    }

    bool CreateListener(int port)
    {
        listener=socket(AF_INET,SOCK_DGRAM,0);
        if(-1==listener)
        {
            std::cout<<"监听文件描述符创建失败! errno is: "<<errno<<std::endl;
            return false;
        }
        struct sockaddr_in addr;
        addr.sin_addr.s_addr=INADDR_ANY;
        addr.sin_family=AF_INET;
        addr.sin_port=htonl(port);
        
        int ret=bind(listener,(sockaddr*)(&addr),sizeof(addr));
        if(ret==-1)
        {
            std::cout<<"绑定监听文件描述符失败! errno is: "<<errno<<std::endl;
            return false;
        }
        return true;
    }

    bool AddEvent(int fd)
    {
        epoll_event event;
        event.data.fd=fd;
        if(fd==listener)
        {
            event.events|=EPOLLIN;
        }
        else 
        {
            event.events|=(EPOLLIN|EPOLLOUT);
        }
        int ret = epoll_ctl(efd,EPOLL_CTL_ADD,fd,&event);
        if(ret==-1)
        {
            std::cout<<"事件添加失败!"<<std::endl;
            return false;
        }

        return true;
    }

    bool DisPactch()
    {
        while(is_running)
        {
            epoll_event event;
            int ret=epoll_wait(efd,&event,1,-1);
            if(ret==-1)
            {
                std::cout<<"epoll wait faild! erron is: "<<errno<<std::endl;
                return false;
            }

            if(event.data.fd==listener)
            {
                sockaddr_in peer;
                socklen_t len=ThreadBufferSize;
                int fd = accept(listener,(sockaddr*)&peer,&len);
                if(fd==-1)
                {
                    std::cout<<"accept failed! errno is: "<<std::endl;
                }   
                epoll_event temp_con;
                temp_con.events|=(EPOLLIN|EPOLLOUT);
                epoll_ctl(efd,EPOLL_CTL_ADD,fd,&temp_con);

                epoll_event temp_listen;
                temp_listen.events|=EPOLLIN;
                epoll_ctl(efd,EPOLL_CTL_ADD,listener,&temp_listen);
            }
            else 
            {
                //让线程池去处理链接操作。TODO  【done!】
                Connection conn;
                conn.socketfd=event.data.fd;
                conn.event=event.events;
                threadpool.AddConnection(conn);
                threadpool.NotifyOneThread();
            }
            
        }
        return true;
    }
};